Skip to content

Pluggable Backend Interface with DataFusion for Bounded-Memory Compute#3716

Open
qzyu999 wants to merge 18 commits into
apache:mainfrom
qzyu999:pluggable-backend-init
Open

Pluggable Backend Interface with DataFusion for Bounded-Memory Compute#3716
qzyu999 wants to merge 18 commits into
apache:mainfrom
qzyu999:pluggable-backend-init

Conversation

@qzyu999

@qzyu999 qzyu999 commented Jul 26, 2026

Copy link
Copy Markdown
Contributor

Which issue does this PR close?

Closes #3715
Closes #1210
Closes #3270
Closes #3554

Rationale for this change

PyIceberg's I/O and compute logic lives in a single 3,000+ line file (pyiceberg/io/pyarrow.py) with no separation of concerns. PyArrow is a kernel library without memory management or spill-to-disk, causing OOM crashes on production-scale operations (CoW deletes, equality delete resolution, scan planning).

This PR introduces a pluggable interface that:

  1. Decouples PyIceberg from PyArrow, enabling alternative backends
  2. Integrates DataFusion for bounded-memory compute with spill-to-disk
  3. Fixes OOM patterns while maintaining full backward compatibility

Design Doc: Support for PyIceberg Pluggable Interface

What changes are included in this PR?

New Modules (pyiceberg/execution/)

Module Purpose
protocol.py ReadBackend, WriteBackend, ComputeBackend protocol definitions
engine.py Backend resolution, instantiation, config helpers
_orchestrate.py Per-task scan execution with delete resolution
planning.py InMemoryPlanner + BoundedMemoryPlanner
backends/pyarrow_backend.py Default implementation (always available)
backends/datafusion_backend.py Bounded-memory implementation (optional)

Operations Delivered

Operation Before After
Equality delete resolution ValueError ✅ Works (Grace Hash Join with spill)
CoW delete (large files) OOM ✅ Two-pass streaming, O(batch_size) memory
Positional deletes (millions) OOM ✅ Size-based routing to anti-join
Sort-on-write Not implemented ✅ External merge sort (best-effort)
Scan planning (>100K deletes) OOM ✅ BoundedMemoryPlanner with SQL join

Migration

All data paths in table/__init__.py now route through orchestrate_scan() instead of ArrowScan. The old ArrowScan class is deprecated (0.11.0) and scheduled for removal in 0.12.0.

Are there any user-facing changes?

No breaking changes. All existing APIs work identically.

New behavior (when datafusion is installed):

  • Tables with equality deletes now return correct results (previously ValueError)
  • Large file operations complete without OOM
  • Files are sorted on write if table defines a sort order

New optional dependency:

pip install 'pyiceberg[datafusion]'

When DataFusion is not installed, behavior is identical to before (may OOM on large data).

New configuration (optional, sensible defaults):

# .pyiceberg.yaml
execution:
  compute-backend: datafusion  # auto-detected when installed
  memory-limit: 536870912      # 512 MB
  planning-threshold: 100000   # switch to bounded planner
  cow-threshold: 67108864      # 64 MB for two-pass streaming

Testing

Includes an exhaustive test suite with a ~5:1 test-to-code LOC ratio:

  • test_arrowscan_parity.py: verifies new path matches deprecated ArrowScan
  • test_property_based.py: Hypothesis tests for PyArrow/DataFusion equivalence
  • test_positional_deletes.py: DataFusion positional delete resolution
  • test_equality_deletes.py: equality delete resolution with sequence number gating
  • test_cow_streaming.py: two-pass streaming for large files
  • All existing integration tests pass without modification

qzyu999 added 7 commits July 25, 2026 10:39
- Fix ruff import sorting (I001) across execution module and tests
- Add type: ignore comments for pyarrow-stubs incompatibilities:
  - Statistics.has_null_count attr-defined (stub missing, runtime exists)
  - ParquetWriter compression Literal type (accepts any string at runtime)
  - binary_join_element_wise overload (stub doesn't match ChunkedArray usage)
  - pa.schema() field list type (stub expects Field[Any] but field() works)
- Add noqa: SIM115 for intentional NamedTemporaryFile(delete=False) pattern
  (needed for temp file paths passed to external code)
- Fix BLE001/S110: Replace broad Exception with specific types in finalizers
  (OSError, RuntimeError for I/O; AttributeError, TypeError for shutdown)
- Fix SIM102: Combine nested if statements in bounds checking
- Fix SIM101: Merge isinstance calls for datetime types
- Update _resolve_filesystem to accept Mapping[str, Any] (matches protocol)
- Add assertions for pa_schema before ParquetWriter (satisfies mypy)
- Filter None values from position delete pylist (satisfies set[int] type)
- Use Any return type for _any_null_mask and composite key builders
  (avoids complex ChunkedArray type param issues with pyarrow-stubs)
- Remove unused noqa: E402 comments in test files
- Remove unused noqa: S301 in planning.py (rule not enabled)

The remaining EXE002 errors in Docker are false positives from Windows volume
mount permissions; git tracks files as 644 and Linux CI won't see them.
- Fix E402: Add noqa comments for imports after pytest.importorskip
- Fix E501: Break long line in test_orchestrate.py
- Remove unused type: ignore comments (CI mypy doesn't need them)
- Add type annotations to test fixtures and test methods
- Fix return type annotations on hypothesis composite strategies
- Fix _null_sentinel duplicate function and return type
- Remove unused type: ignore comments in pyiceberg/transforms.py and pyiceberg/avro/file.py
- Add missing return statements in transforms.py project/strict_project methods
- Add Path imports and type annotations to test fixtures
- Fix line length issues in test files
- Add missing imports (Any, InMemoryCatalog, ComputeBackend) to test files

Source files now pass mypy. Test files have partial type coverage.
qzyu999 added 2 commits July 26, 2026 22:00
- Add proper type annotations to fixtures and test functions
- Fix Generator return types for contextmanagers
- Fix Iterator imports and annotations
- Add type: ignore comments for legitimate edge cases (testing Mapping immutability, unreachable after pytest.raises)
- Fix SortField transform parameter to use IdentityTransform() instead of string
- Remove unused type: ignore comments
Reverts incorrect removal of type: ignore comments that are required
for these files to pass mypy in the existing codebase.
@0guban0v

Copy link
Copy Markdown
image

qzyu999 added 9 commits July 26, 2026 22:20
The previous condition only triggered reconciliation when field IDs differed,
but this missed cases where nullability differed between the Parquet file
schema and the Iceberg table schema.

For example, a Parquet file written with nullable=False should be cast to
nullable=True when the Iceberg schema specifies required=False (optional).

The original ArrowScan.to_record_batches unconditionally called
_to_requested_schema for every batch. This fix restores that behavior.

Fixes: test_add_file_with_valid_nullability_diff
1. Bind residual predicates before passing to compute.filter()
   - ResidualEvaluator can return unbound predicates for unpartitioned tables
   - The original ArrowScan bound the row filter; orchestrate_scan must do the same
   - Fixes: test_partitioned_table_delete_full_file

2. Use parsed path instead of full URI for local filesystem
   - _resolve_filesystem was passing the full 'file:/path' URI to os.path.abspath
   - Should use the parsed 'path' component to strip the scheme prefix
   - Fixes: test_create_table_transaction[memory-1]
1. Schema reconciliation: Only reconcile when actually needed
   - Trigger on field ID differences (schema evolution)
   - Trigger when file is required but projected is optional
   - Skip when only types differ (inferred Arrow types may not match Iceberg)
   - This fixes test failures from type promotion errors (long  int)

2. Predicate binding: Handle already-bound predicates gracefully
   - Some residuals (from tests or REST planning) are pre-bound
   - Catch TypeError from bind() and use the predicate as-is
   - This fixes test failures from re-binding bound predicates
1. Add restore_transaction_class_properties autouse fixture to
   tests/execution/conftest.py to restore Transaction.table_metadata
   after tests that use PropertyMock. This prevents class-level mocks
   from leaking into subsequent tests (fixes test_add_top_level_primitives
   failure on Python 3.14).

2. Fix _infer_file_schema_from_batch to use table_metadata.name_mapping()
   (stored name mapping with old aliases) instead of schema().name_mapping
   (current names only). This enables proper schema reconciliation when
   reading files written before column renames.
…ead pushdown

This fixes two test failures:
1. test_upsert_with_identifier_fields - upsert scan wasn't finding all rows
2. test_partitioned_table_rewrite - delete was removing all rows

The root cause was that orchestrate_scan used task.residual for read pushdown
but row_filter for post-filtering. When _read_live_rows passed row_filter=AlwaysTrue()
to read all rows (for CoW delete), the pushdown still used task.residual (which
contained the delete filter), causing only rows matching the delete filter to be read.

Now both pushdown and post-filter consistently use the row_filter parameter,
which allows callers to override the task's residual when needed.
1. Fix test_scan_calls_filter_for_residual:
   - Changed test to pass row_filter=bound_filter instead of relying on task.residual
   - This matches the new orchestrate_scan behavior where post-filter uses row_filter

2. Fix pushdown filter logic:
   - Use task.residual for pushdown (handles schema evolution column names)
   - When row_filter is AlwaysTrue, use AlwaysTrue for pushdown too
   - This fixes _read_live_rows for CoW delete where we need all rows

3. Fix test_delete_after_partition_evolution_from_unpartitioned:
   - Added check for column name differences in _build_reconcile_fn
   - When file has 'idx' but projected schema has 'id', trigger reconciliation
   - This ensures batches are renamed to match current schema before filtering
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

2 participants